package com.atguigu.Flink.datastream.transform;

import com.atguigu.Flink.POJO.WordCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import javax.lang.model.element.VariableElement;

//* 富函数:  RichFunction
//        *    基本上每个函数都有对应的富函数， 例如;
//        *       FlatMapFunction  => RichFlatMapFunction
//        *       MapFunction    => RichMapFunction
//        *       FilterFunction  => RichFilterFunction
//        *
//        * 富函数的功能:
//        *    1. 算子生命周期方法
//        *         open() :  算子初始化的时候调用一次
//        *         close():  算子结束的时候调用一次。
//        *
//        *    2. 可以获取运行时上下文对象 RuntimeContext ，进而获取更多作业相关详细的信息
//        *       最重要的就是可以做状态编程。
public class Flink05_RichFunction {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        FileSource<String> fileSource
                = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("Input/clicks.txt")).build();

        DataStreamSource<String> ds = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "rc");

        ds.flatMap(new RichFlatMapFunction<String, WordCount>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("算子初始化");
            }

            @Override
            public void close() throws Exception {
                System.out.println("算子结束");
            }

            @Override
            public void flatMap(String s, Collector<WordCount> collector) throws Exception {
                RuntimeContext runtimeContext = getRuntimeContext();
                runtimeContext.getTaskName();
                System.out.println(runtimeContext.getJobId());
                runtimeContext.getIndexOfThisSubtask();
                String[] value = s.split(",");
                collector.collect(new WordCount(value[0],1L));
            }
        }).keyBy(WordCount::getWord).sum("count").print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
